package com.lmf.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 手动实现一个分布式锁的案例：
 */
public class DistributedLock {
    private String connectString = "192.168.120.128:2181,192.168.120.130:2181,192.168.120.131:2181";
    private int sessionTimeOut = 200000;
    private ZooKeeper zk;
    private CountDownLatch connectLatch = new CountDownLatch(1);//创建一个倒计时锁存器
    private String waitPath;//前一个顺序节点的路径
    private CountDownLatch waitLatch = new CountDownLatch(1);//创建一个倒计时锁存器
    private String currentMode;



    //构造方法
    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        //1.获取zk连接
        zk = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // connectLatch  如果连接上zk  可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    connectLatch.countDown();
                }
                // waitLatch  需要释放
                if (watchedEvent.getType()== Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });

        //1.1 等待zk连接成功，程序再继续往下执行
        connectLatch.await();//必须要手动释放之后，才可以执行

        //1.2 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);
        if (stat == null) {
            //若不存在，则创建根节点
            zk.create("/locks","locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }

    }

    //2.对zk加锁：判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点
    public void zkLock(){
        try {
        // 加锁：即，创建对应的临时带序号节点
            currentMode = zk.create("/locks/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // wait一小会, 让结果更清晰一些
            Thread.sleep(10);
            //获取锁下面的所有节点
            List<String> children = zk.getChildren("/locks", false);//不需要监视器
            // 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小
            if (children.size() == 1) {
                return;
            }else {
                //集合自然排序
                Collections.sort(children);
                // 获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过seq-00000000获取该节点在children集合的位置
                int index = children.indexOf(thisNode);
                if (index == -1) {
                    System.out.println("数据有误！");
                }else if (index == 0){
                    //节点序号最小，可以继续执行
                    return;
                }else {
                    //不是序号最小的节点，则监视前一个节点
                    //前一个节点路径
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath, true, new Stat());//需要监视器
                    // 等待监听前一个节点被删除，才能继续执行
                    // 必须等监听结束，才能继续往下执行
                    waitLatch.await();//运用锁，来限制程序执行
                    return;
                }

            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }



    //3.对zk解锁
    public void unZkLock(){
        //解锁：删除节点
        try {
            zk.delete(this.currentMode,-1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}

